#!/usr/bin/env python
# -*- encoding: utf-8 -*-
# Project: spd-sxmcc
"""
@author: lyndon
@time Created on 2019/1/6 6:39
@desc
"""
import Queue

from pykafka import KafkaClient

# 是因为kafka传递的字节，因此在传递字符串处encode()即可，分别是client.topics和producer.produce()，如下
host = '192.168.100.151:9092,192.168.100.152:9092,192.168.100.153:9092'
client = KafkaClient(hosts=host)



def sync_msg_topic():
    '''
    将产生kafka同步消息，这个调用仅仅在我们已经确认消息已经发送到集群之后
    :return:
    '''
    topic = client.topics["test".encode()]
    with topic.get_sync_producer() as producer:
        for i in range(100):
            producer.produce(('test message ' + str(i ** 2)).encode())



# 可接受多个client
# 查看所有的topic
print (client.topics)

topic = client.topics['test']  # 选择一个topic
message = "test message test message"
# # 当有了topic之后呢，可以创建一个producer,来发消息，生产kafka数据,通过字符串形式，
# with topic.get_sync_producer() as producer:
#     producer.produce(message)
# 以上的例子将产生kafka同步消息，这个调用仅仅在我们已经确认消息已经发送到集群之后

# 但生产环境，为了达到高吞吐量，要采用异步的方式，通过delivery_reports =True来启用队列接口；
producer = topic.get_producer(sync=False, delivery_reports=True)
producer.produce(message.encode())
try:
    msg, exc = producer.get_delivery_report(block=False)
    if exc is not None:
        print('Failed to deliver msg {}: {}'.format(msg.partition_key, repr(exc)))
    else:
        print('Successfully delivered msg {}'.format(msg.partition_key))
except Queue.Empty:
    pass
